package cn.doitedu.hbase.baseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

/**
 * @date: 2019/7/11
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description: Hbase的数据查询api示例代码
 */
public class QueryDemos {

    Connection conn = null;
    TableName t1 = TableName.valueOf("t1");


    @Before
    public void init() throws Exception {

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "spark01:2181,spark02:2181,spark03:2181");

        conn = ConnectionFactory.createConnection(conf);
    }


    /**
     * get数据的api示例
     * @throws IOException
     */
    @Test
    public void testGet() throws IOException {


        Table table = conn.getTable(t1);

        Get get = new Get("z009".getBytes());
        // 过滤要返回的数据：满足条件的才返回
        get.setFilter(null);

        // Result就是对一行中的kv数据的封装
        Result result = table.get(get);

        // 知道数据schema的情况下，直接取某个key的value
        /*byte[] value = result.getValue("base_info".getBytes(), "age".getBytes());
        System.out.println(new String(value));*/


        // 如果每行的schama不一致，也可以从result中遍历出每一个key-value
        CellScanner cellScanner = result.cellScanner();
        while (result.advance()){
            Cell cell = result.current();
            printCellUtil(cell);

        }

        table.close();
        conn.close();
    }


    /**
     * 一次RPC请求，get多行数据
     * @throws IOException
     */
    @Test
    public  void testGetSome() throws IOException {

        Table table = conn.getTable(t1);

        Get get9 = new Get("z009".getBytes());
        Get get10 = new Get("z0010".getBytes());
        Get get11 = new Get("z0011".getBytes());
        Get get12 = new Get("z0012".getBytes());

        ArrayList<Get> gets = new ArrayList<>();
        gets.add(get9);
        gets.add(get10);
        gets.add(get11);
        gets.add(get12);

        Result[] results = table.get(gets);
        for (Result result : results) {
            while(result.advance()){
                Cell cell = result.current();
                printCellUtil(cell);
            }
        }

        table.close();
        conn.close();
    }


    /**
     * 范围扫描 scan  示例代码
     * @throws IOException
     */
    @Test
    public void testScan() throws IOException {

        Table table = conn.getTable(t1);


        Scan scan = new Scan();
        // 指定起始行键
        scan.withStartRow("p001".getBytes());
        // 指定扫描的结束行键
        scan.withStopRow("p003".getBytes(),true);
        // 指定返回数据只包含 base_info 列族的kv
        scan.addFamily("base_info".getBytes());

        // 设置返回结果的数据过滤器（相当于实现条件查询）
        scan.setFilter(null);

        // 设置在本scan中，一个Result中最多包含的kv个数；用于一行数据中kv量太大的情况
        // 假如一行中有如下kv:      k1v1,k2v2,k3v3,k4v4,k5v5,k6v6,k7v7,k8v8
        // 那么，这一行数据返回的result就有3个
        scan.setBatch(3);

        // 是否让regionserver侧缓存本次扫描到的数据到内存中
        scan.setCacheBlocks(true);


        // 指定本次scan最多返回的result个数
        scan.setLimit(10);


        // 指定本次scan按反方向进行，从stoprowkey->startrowkey
        scan.setReversed(false);

        // 从另一个维度：字节大小  来限制一个result的大小
        scan.setMaxResultSize(10240);

        // 指定本次scan是否要包含raw数据（包含已经被删除的，或者过时的版本数据）
        scan.setRaw(true);


        // 指定返回数据只需要base_info列族的
        //scan.addFamily("base_info".getBytes());


        ResultScanner rsScanner = table.getScanner(scan);
        Iterator<Result> iter = rsScanner.iterator();


        // 迭代每一个result
        while(iter.hasNext()){
            Result rs = iter.next();
            // 清楚schema的状态下，可以直捣黄龙，取value
            /*rs.getValue("base_info".getBytes(),"name".getBytes());
            rs.getValue("base_info".getBytes(),"age".getBytes());*/

            // 不清楚schema或者每一行的schema不一致的情况下，可以摸黑去遍历
            // 迭代result中的每一个cell
            while(rs.advance()){
                Cell cell = rs.current();
                printCellUtil(cell);
            }
        }

        rsScanner.close();
        table.close();
        conn.close();


    }


    /**
     * cell数据遍历的工具写法
     * @param cell
     */
    public static void printCellUtil(Cell cell){

        byte[] r = CellUtil.cloneRow(cell);
        byte[] f = CellUtil.cloneFamily(cell);
        byte[] q = CellUtil.cloneQualifier(cell);
        byte[] v = CellUtil.cloneValue(cell);


        String rs = new String(r);
        String fs = new String(f);
        String qs = new String(q);
        String vs = new String(v);
        // 如果value是别的类型，可以用Bytes工具来解码
        // int i = Bytes.toInt(v);

        System.out.println(rs+ " -> " + fs +" -> " +qs +" -> " + vs );

    }


    /**
     * cell遍历的底层写法
     * @param cell
     */
    public static void printCell(Cell cell){
        // rowArray是整条数据的bytes
        byte[] rowArray = cell.getRowArray();
        int rowOffset = cell.getRowOffset();
        short rowLength = cell.getRowLength();


        byte[] familyArray = cell.getFamilyArray();
        int familyOffset = cell.getFamilyOffset();
        byte familyLength = cell.getFamilyLength();

        byte[] qualifierArray = cell.getQualifierArray();
        int qualifierOffset = cell.getQualifierOffset();
        int qualifierLength = cell.getQualifierLength();


        byte[] valueArray = cell.getValueArray();
        int valueOffset = cell.getValueOffset();
        int valueLength = cell.getValueLength();


        String r = new String(rowArray,rowOffset,rowLength);
        String f = new String(familyArray,familyOffset,familyLength);
        String q = new String(qualifierArray,qualifierOffset,qualifierLength);
        String v = new String(valueArray,valueOffset,valueLength);

        System.out.println(r+ " -> " + f +" -> " +q +" -> " + v );
    }



}
